﻿namespace Cellfish.Toolbox.TaskSchedulers
{
    using System;
    using System.Collections.Generic;
    using System.Linq;
    using System.Threading;
    using System.Threading.Tasks;

    using Cellfish.Toolbox.Collections.Concurrent;

    /// <summary>
    /// Represents an object that handles the low-level work of queuing tasks onto a private pool of threads.
    /// </summary>
    public class PrivateThreadPoolTaskScheduler : TaskScheduler, IDisposable
    {
        private static readonly int cpuCount = Math.Max(Environment.ProcessorCount, 1);

        private const int MaxThreadsPerCpu = 100;

        private Thread[] threads;

        private volatile bool isShutdown = false;

        private readonly ConcurrentLinkedList<Task> taskQueue = new ConcurrentLinkedList<Task>();

        private ManualResetEvent queueSignal = new ManualResetEvent(false);

        /// <summary>
        /// Initializes a new instance of the <see cref="PrivateThreadPoolTaskScheduler"/> class.
        /// </summary>
        /// <remarks>The thread pool will have one thread per CPU.</remarks>
        public PrivateThreadPoolTaskScheduler()
            : this(cpuCount)
        {
        }

        /// <summary>
        /// Initializes a new instance of the <see cref="PrivateThreadPoolTaskScheduler"/> class.
        /// </summary>
        /// <param name="numberOfThreads">The number of threads in the thread pool.</param>
        /// <exception cref="System.ArgumentOutOfRangeException">If the number of threads requested is too large.</exception>
        public PrivateThreadPoolTaskScheduler(int numberOfThreads)
        {
            if (numberOfThreads <= 0 || numberOfThreads >= (cpuCount * MaxThreadsPerCpu))
            {
                throw new ArgumentOutOfRangeException("numberOfThreads", string.Format("The number of threads must be between 1 and {0}", cpuCount * MaxThreadsPerCpu));
            }

            this.threads = new Thread[numberOfThreads];
            for (int i = 0; i < this.threads.Length; i++)
            {
                this.threads[i] = new Thread(this.ThreadBody);
                this.threads[i].Start();
            }
        }

        /// <summary>
        /// Shutdowns the scheduler trying to exit all worker threads.
        /// Will wait up until given timeout for worker threads to complete.
        /// </summary>
        /// <param name="timeout">The timeout.</param>
        public void Shutdown(TimeSpan timeout)
        {
            if (!this.isShutdown)
            {
                this.isShutdown = true;
                this.queueSignal.Set();
                var utcTimeout = DateTime.UtcNow + timeout;
                foreach (var thread in this.threads)
                {
                    var threadTimeout = utcTimeout - DateTime.UtcNow;
                    if (threadTimeout >= TimeSpan.Zero)
                    {
                        thread.Join(threadTimeout);
                    }
                    else
                    {
                        thread.Join(TimeSpan.Zero);
                    }
                }
            }
        }

        /// <summary>
        /// Indicates the maximum concurrency level this <see cref="T:System.Threading.Tasks.TaskScheduler" /> is able to support.
        /// </summary>
        /// <returns>Returns an integer that represents the maximum concurrency level. The default scheduler returns <see cref="F:System.Int32.MaxValue" />.</returns>
        public sealed override int MaximumConcurrencyLevel
        {
            get
            {
                return this.threads.Length;
            }
        }

        /// <summary>
        /// Queues a <see cref="T:System.Threading.Tasks.Task" /> to the scheduler.
        /// </summary>
        /// <param name="task">The <see cref="T:System.Threading.Tasks.Task" /> to be queued.</param>
        /// <exception cref="System.Threading.Tasks.TaskSchedulerException">Scheduler is shutting down.</exception>
        protected sealed override void QueueTask(Task task)
        {
            if (this.isShutdown)
            {
                throw new TaskSchedulerException("Scheduler is shutting down.");
            }

            this.taskQueue.AddLast(task);
            this.queueSignal.Set();
        }

        /// <summary>
        /// Determines whether the provided <see cref="T:System.Threading.Tasks.Task" /> can be executed synchronously in this call, and if it can, executes it.
        /// </summary>
        /// <param name="task">The <see cref="T:System.Threading.Tasks.Task" /> to be executed.</param>
        /// <param name="taskWasPreviouslyQueued">A Boolean denoting whether or not task has previously been queued. If this parameter is True, then the task may have been previously queued (scheduled); if False, then the task is known not to have been queued, and this call is being made in order to execute the task inline without queuing it.</param>
        /// <returns>
        /// A Boolean value indicating whether the task was executed inline.
        /// </returns>
        protected sealed override bool TryExecuteTaskInline(Task task, bool taskWasPreviouslyQueued)
        {
            if (this.IsPrivateThreadPoolThread && !this.isShutdown)
            {
                if (!taskWasPreviouslyQueued || this.TryDequeue(task))
                {
                    return base.TryExecuteTask(task);
                }
            }

            return false;
        }

        /// <summary>
        /// For debugger support only, generates an enumerable of <see cref="T:System.Threading.Tasks.Task" /> instances currently queued to the scheduler waiting to be executed.
        /// </summary>
        /// <returns>
        /// An enumerable that allows a debugger to traverse the tasks currently queued to this scheduler.
        /// </returns>
        protected override IEnumerable<Task> GetScheduledTasks()
        {
            return this.taskQueue.ToArray();
        }

        /// <summary>
        /// Attempts to dequeue a <see cref="T:System.Threading.Tasks.Task" /> that was previously queued to this scheduler.
        /// </summary>
        /// <param name="task">The <see cref="T:System.Threading.Tasks.Task" /> to be dequeued.</param>
        /// <returns>
        /// A Boolean denoting whether the <paramref name="task" /> argument was successfully dequeued.
        /// </returns>
        protected sealed override bool TryDequeue(Task task)
        {
            return this.taskQueue.Remove(task);
        }

        private void ThreadBody()
        {
            while (!this.isShutdown)
            {
                Task task = null;
                if (!this.taskQueue.TryTake(out task))
                {
                    if (this.IsQueueSignalOn)
                    {
                        this.queueSignal.Reset();
                        if (this.taskQueue.Count > 0)
                        {
                            this.queueSignal.Set();
                        }
                    }
                    
                    this.queueSignal.WaitOne();
                }

                if (task != null)
                {
                    base.TryExecuteTask(task);
                }
            }
        }

        private bool IsQueueSignalOn
        {
            get
            {
                return this.queueSignal.WaitOne(TimeSpan.Zero);
            }
        }

        private bool IsPrivateThreadPoolThread
        {
            get
            {
                return this.threads.Any(t => t.ManagedThreadId == Thread.CurrentThread.ManagedThreadId);
            }
        }
        #region IDisposable
        /// <summary>
        /// Performs application-defined tasks associated with freeing, releasing, or resetting unmanaged resources.
        /// </summary>
        public void Dispose()
        {
            this.Dispose(true);
            GC.SuppressFinalize(this);
        }

        /// <summary>
        /// Releases unmanaged and - optionally - managed resources.
        /// </summary>
        /// <param name="disposing"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
        protected void Dispose(bool disposing)
        {
            if (disposing)
            {
                this.Shutdown(TimeSpan.Zero);
                this.queueSignal.Dispose();
            }
        }
        #endregion
    }
}
